package com.xiaochong.bitong.api.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.bitong.api.config.hbase.HbaseConnectionFactory;
import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseTablePutExcetion;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Project : xc-tir
 * @Package Name : com.xiaochong.bitong.api.component
 * @Description : TODO
 * @Author : daizhiqing@xiaochong.com
 * @Creation Date : 2018年08月01日上午10:54
 * @ModificationHistory Who When What
 */
@Component
public class HbaseComponent {


    @Autowired
    private HbaseConnectionFactory hbaseConnectionFactory;


    /**
     * 读取hbase 数据
     * @param tableName
     * @param rowKey
     * @return param table ，rowKey
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public  Map<String,Object> selectData(String tableName,String rowKey) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Map<String,Object> paramMap = new HashMap<String,Object>();
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            String qualifier = new String(CellUtil.cloneQualifier(cell),"UTF-8");
            String value = new String(CellUtil.cloneValue(cell),"UTF-8");
            if(StringUtils.isNotBlank(value) && !value.equals("null")){
                paramMap.put(qualifier,value);
            }

        }
        hbaseConnectionFactory.closeAll(conn,table);
        return paramMap;
    }

    /**
     * 读取hbase 数据
     * @param tableName
     * @param rowKey
     * @return param key 为列名，value 为值 ,timeStamp
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public Map<String,Object> selectDataByTimeStamp(String tableName, String rowKey, Long timeStamp) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Map<String,Object> paramMap = new HashMap<String,Object>();
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setTimeStamp(timeStamp);
        Result result = table.get(get);
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            String qualifier = new String(CellUtil.cloneQualifier(cell),"UTF-8");
            String value = new String(CellUtil.cloneValue(cell),"UTF-8");
            if(StringUtils.isNotBlank(value) && !value.equals("null")){
                paramMap.put(qualifier,value);
            }

        }
        hbaseConnectionFactory.closeAll(conn,table);
        return paramMap;
    }

    /**
     * 指定列族列明查询
     * @param tableName
     * @param rowKey
     * @return param key 为列名，value 为值
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     */
    public  String selectByQualifierData(String tableName,String family,String qualifier ,String rowKey) throws HbaseConnectionExcetion, HbaseGetTableExcetion, IOException, HbaseCloseExcetion {
        Connection conn  = hbaseConnectionFactory.getConnection();
        Table table= hbaseConnectionFactory.getTable(conn,tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        Result result = table.get(get);
        String resultValue = null;
        Cell cells[]= result.rawCells();
        for(Cell cell : cells){
            resultValue = new String(CellUtil.cloneValue(cell),"UTF-8");
        }
        hbaseConnectionFactory.closeAll(conn,table);

        return resultValue;
    }

    /**
     * 插入hbase数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @param value
     */
    public void insertHbaseString(String tableName, String rowKey, String family, String qualifier, String value){
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            Put put = new Put(rowKey.getBytes());
            put.addColumn(Bytes.toBytes(family),
                    Bytes.toBytes(qualifier),
                    Bytes.toBytes(value));
            hbaseConnectionFactory.tablePut(table,put);
            hbaseConnectionFactory.closeAll(connection,table);

        } catch (HbaseConnectionExcetion | HbaseGetTableExcetion | HbaseCloseExcetion | HbaseTablePutExcetion e) {
            LogTrace.error(e.getMessage(), e);
        }
    }


    /**
     * 插入hbase数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param object
     */
    public void insertHbaseObject(String tableName, String rowKey, String family,Object object) {
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            Put put = new Put(rowKey.getBytes());
            JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(object));
            Map<String, Object> map = jsonObject.getInnerMap();
            for(Map.Entry<String, Object> entry: map.entrySet()){
                String key = entry.getKey();
                put.addColumn(Bytes.toBytes(family),
                        Bytes.toBytes(key),
                        Bytes.toBytes(entry.getValue()!=null?entry.getValue().toString():"") );
            }
            hbaseConnectionFactory.tablePut(table,put);
            hbaseConnectionFactory.closeAll(connection,table);
        } catch (HbaseCloseExcetion | HbaseGetTableExcetion | HbaseTablePutExcetion | HbaseConnectionExcetion e) {
            e.printStackTrace();
            LogTrace.error(e.getMessage(),e);
        }
    }



    /**
     * hbase批量插入数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param list
     */
    public void insertHbaseList(String tableName, String rowKey, String family,List list) {
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            List<Put> putList = new ArrayList<>();
            for (int i = 0; i < list.size(); i++) {
                Put put = new Put(rowKey.getBytes());//row kye
                JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(list.get(i)));
                Map<String, Object> map = jsonObject.getInnerMap();
                for(Map.Entry<String, Object> entry: map.entrySet()){
                    String key = entry.getKey();
                    put.addColumn(Bytes.toBytes(family),
                            Bytes.toBytes(key),
                            Bytes.toBytes(entry.getValue()!=null?entry.getValue().toString():"") );
                }

                putList.add(put);
            }
            hbaseConnectionFactory.tablePut(table,putList);
            hbaseConnectionFactory.closeAll(connection,table);
        } catch (HbaseCloseExcetion | HbaseGetTableExcetion | HbaseTablePutExcetion | HbaseConnectionExcetion e) {
            e.printStackTrace();
            LogTrace.error(e.getMessage(),e);
        }
    }

}
